﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Xml;
using System.IO;
using System.Security.Policy;
using System.Runtime.Remoting;
using System.Runtime.Remoting.Lifetime;


namespace CompressedFileDest
{
     [DtsPipelineComponent
         (
         DisplayName = "CompressedFileDest", Description = "Creator of a compressed file straight from the pipeline",
         ComponentType = ComponentType.DestinationAdapter
         )
    ]
    public class CompressedFileDest : PipelineComponent 
    {

          public override void PreExecute()
         {
             base.PreExecute();
             /*
              * Add your code here
              */
             rowsInCurrentFile = 0;
             curFile = 0;
             tableName = @"TableData.dat";


             columnGetters = new DataGetter[ComponentMetaData.InputCollection[0].InputColumnCollection.Count];
             Connection = ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager;
             int i = 0;
             foreach (IDTSInputColumn100 column in ComponentMetaData.InputCollection[0].InputColumnCollection)
             {
                 if (column.DataType == DataType.DT_BOOL)
                     columnGetters[i] = new BooleanGetter();
                 else if (column.DataType == DataType.DT_BYTES)
                     columnGetters[i] = new BytesGetter();
                 else if (column.DataType == DataType.DT_CY || column.DataType == DataType.DT_DECIMAL)
                     columnGetters[i] = new DecimalGetter();
                 else if (column.DataType == DataType.DT_DATE)
                     columnGetters[i] = new DateTimeGetter();
                 else if (column.DataType == DataType.DT_I8)
                     columnGetters[i] = new Int64Getter();
                 else if (column.DataType == DataType.DT_BYTES)
                     columnGetters[i] = new BytesGetter();
                 else if (column.DataType == DataType.DT_DBDATE)
                     columnGetters[i] = new DateGetter();
                 else if (column.DataType == DataType.DT_DBTIME)
                     columnGetters[i] = new TimeGetter();
                 else if (column.DataType == DataType.DT_DBTIME2)
                     columnGetters[i] = new TimeGetter();
                 else if (column.DataType == DataType.DT_DBTIMESTAMP || column.DataType == DataType.DT_DBTIMESTAMP2)
                     columnGetters[i] = new DateTimeGetter();
                 else if (column.DataType == DataType.DT_DBTIMESTAMPOFFSET)
                     columnGetters[i] = new DateTimeOffsetGetter();
                 else if (column.DataType == DataType.DT_FILETIME)
                     columnGetters[i] = new DateTimeGetter();
                 else if (column.DataType == DataType.DT_GUID)
                     columnGetters[i] = new GuidGetter();
                 else if (column.DataType == DataType.DT_I1 || column.DataType == DataType.DT_UI1)
                     columnGetters[i] = new ByteGetter();
                 else if (column.DataType == DataType.DT_I2)
                     columnGetters[i] = new Int16Getter();
                 else if (column.DataType == DataType.DT_I4)
                     columnGetters[i] = new Int32Getter();
                 else if (column.DataType == DataType.DT_I8)
                     columnGetters[i] = new Int64Getter();
                 else if (column.DataType == DataType.DT_NUMERIC)
                     columnGetters[i] = new DecimalGetter();
                 else if (column.DataType == DataType.DT_R4)
                     columnGetters[i] = new SingleGetter();
                 else if (column.DataType == DataType.DT_R8)
                     columnGetters[i] = new DoubleGetter();
                 else if (column.DataType == DataType.DT_BYTES)
                     columnGetters[i] = new BytesGetter();
                 else if (column.DataType == DataType.DT_BYTES)
                     columnGetters[i] = new SingleGetter();
                 else if (column.DataType == DataType.DT_STR || column.DataType == DataType.DT_WSTR)
                     columnGetters[i] = new StringGetter();
                 else if (column.DataType == DataType.DT_UI2)
                     columnGetters[i] = new UInt16Getter();
                 else if (column.DataType == DataType.DT_UI4)
                     columnGetters[i] = new UInt32Getter();
                 else if (column.DataType == DataType.DT_UI8)
                     columnGetters[i] = new UInt64Getter();
                 else if (column.DataType == DataType.DT_IMAGE)
                     columnGetters[i] = new ImageGetter();
                 else if (column.DataType ==  DataType.DT_TEXT || column.DataType == DataType.DT_NTEXT)
                     columnGetters[i] = new TextGetter();
                 else columnGetters[i] = new BytesGetter();



                 // instantiate the right kind:
                 columnGetters[i].colIndex = i;
                 columnGetters[i].colLength = column.Length;
                 columnGetters[i].colName = column.Name;
                 columnGetters[i].precision = column.Precision;
                 columnGetters[i].scale = column.Scale;
                 columnGetters[i].parent = this;
                 i++;
             }

 
         }

         public override void PostExecute()
         {
             base.PostExecute();

             // need to lookup the table name here:
             //ComponentMetaData.FireInformation(10, ComponentMetaData.Name, "Closing file " + this.Connections.Connection.ConnectionString.ToString() + ((curFile == 1) ? "" : (curFile - 1).ToString()), "", 0, false);
             if (writeStream != null)
                 writeStream.Close();
             if (dataFile != null)
                 dataFile.Close();

         }

         public override void ProcessInput(int inputID, PipelineBuffer buffer)
         {
             //base.ProcessInput(inputID, buffer);
              PipelineBuffer Row = buffer;
             // first thing, check to see if we need to initialize a file:
             
             if (rowsInCurrentFile == 0 && curFile == 0)
             {
                dataFile = System.IO.Packaging.Package.Open(Connection.ConnectionString.ToString() + ((curFile == 0) ? "" : curFile.ToString().Trim()),
                     System.IO.FileMode.Create);


                 // open the write stream for the Data:
                 Uri partUriDocument = System.IO.Packaging.PackUriHelper.CreatePartUri(
                new Uri(@"/" + tableName, UriKind.Relative));
                 dataPart = dataFile.CreatePart(partUriDocument
                                     , System.Net.Mime.MediaTypeNames.Application.Octet
                                     , System.IO.Packaging.CompressionOption.Normal);
                 WriteMetaDataToFile();
                 writeStream = new System.IO.BinaryWriter(dataPart.GetStream());

                 for (int i = 0; i < columnGetters.Length; i++)
                 {
                     columnGetters[i].processNewFile(ref dataFile, ref writeStream);
                 }
                 // reset the row counter.
                 rowsInCurrentFile = 0;
             }

             // output each column here:
             while (buffer.NextRow())
             {
                 for (int i = 0; i < columnGetters.Length; i++)
                 {
                     columnGetters[i].ProcessData(buffer);
                 }
                 rowsInCurrentFile++;
                 if ((rowsInCurrentFile % 100000) == 0)
                     writeStream.Flush();
             }

         }

         


         public void WriteMetaDataToFile()
         {
             // create the metadat if it doesn't alread exist
             if (dom == null)
             {
                 ComponentMetaData.FireInformation(0, ComponentMetaData.Name, "Creating destination file metadata.", "", 0, true);
                 dom = new XmlDocument();
                 XmlNode xmlnode = dom.CreateNode(XmlNodeType.XmlDeclaration, "", "");
                 dom.AppendChild(xmlnode);
                 // next add the root element:
                 XmlElement elem = dom.CreateElement("", "ROOT", "");
                 dom.AppendChild(elem);
                 XmlNode columns = dom.CreateElement("Columns");

                 IDTSVirtualInputColumnCollection100 cs = this.ComponentMetaData.InputCollection[0].GetVirtualInput().VirtualInputColumnCollection;
                 for (int i = 0; i < cs.Count; i++)
                 {
                     // Record the relevant metadata to read the data at the source
                     XmlNode columnNode = dom.CreateElement("Column");
                     XmlAttribute colAttr = dom.CreateAttribute("Name");
                     colAttr.InnerText = cs[i].Name.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("CodePage");
                     colAttr.InnerText = cs[i].CodePage.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("DataType");
                     colAttr.InnerText = cs[i].DataType.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("Length");
                     colAttr.InnerText = cs[i].Length.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("Precision");
                     colAttr.InnerText = cs[i].Precision.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("Scale");
                     colAttr.InnerText = cs[i].Scale.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("SortKeyPosition");
                     colAttr.InnerText = cs[i].SortKeyPosition.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("UsageType");
                     colAttr.InnerText = cs[i].UsageType.ToString();
                     columnNode.Attributes.Append(colAttr);
                     colAttr = dom.CreateAttribute("Ordinal");
                     colAttr.InnerText = i.ToString();
                     columnNode.Attributes.Append(colAttr);
                     columnNode.Attributes.Append(colAttr);
                     columns.AppendChild(columnNode);
                 }
                 elem.AppendChild(columns);
             }

             Uri partUriDocument = System.IO.Packaging.PackUriHelper.CreatePartUri(
                 new Uri(@"/metaData.xml", UriKind.Relative));
             System.IO.Packaging.PackagePart packagePartDocument =
                 dataFile.CreatePart(partUriDocument,
                 System.Net.Mime.MediaTypeNames.Text.Xml);
             using (System.IO.Stream xmlStream = packagePartDocument.GetStream())
             {
                 XmlWriter w = new XmlTextWriter(xmlStream, System.Text.Encoding.UTF8);
                 dom.WriteContentTo(w);
                 w.Close();
                 xmlStream.Close();
             }

         }

         public override void ProvideComponentProperties()
         {
             // Reset the component.
             base.RemoveAllInputsOutputsAndCustomProperties();
             ComponentMetaData.RuntimeConnectionCollection.RemoveAll();

             ComponentMetaData.InputCollection.New();
             ComponentMetaData.InputCollection[0].Name = "input";
             ComponentMetaData.InputCollection[0].HasSideEffects = true;

             ComponentMetaData.InputCollection[0].ExternalMetadataColumnCollection.New();
             ComponentMetaData.InputCollection[0].ExternalMetadataColumnCollection[0].Name = "External Columns";


             if (ComponentMetaData.RuntimeConnectionCollection.Count == 0)
             {
                 ComponentMetaData.RuntimeConnectionCollection.New();
             }

             ComponentMetaData.RuntimeConnectionCollection[0].Name = "CompressedFileDestOutput";

         }
         public override void AcquireConnections(object Transaction)
         {
             if (ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager != null)
             {
                 Microsoft.SqlServer.Dts.Runtime.ConnectionManager conn = Microsoft.SqlServer.Dts.Runtime.DtsConvert.GetWrapper(
                     ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager);
                 ConnectionManagerFile cmf = conn.InnerObject as ConnectionManagerFile;
                 if (cmf == null)
                    throw new Exception ("The Connection Manager " + conn.Name + "is not a FILE connection");

                 // now make the connection:

                 cmf.AcquireConnection(Transaction);
                 this.Connection = cmf;

                 if (this.Connection == null)
                     throw new Exception ("The Connection Manager " + conn.Name + "is not a FILE connection");
             }

         }

         public override void ReleaseConnections()
         {
             base.ReleaseConnections();
             if (writeStream != null)
                 writeStream.Close();
             if (dataFile != null)
                 dataFile.Close();
             if (ComponentMetaData.RuntimeConnectionCollection[0] != null && ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager != null)
             ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager.ReleaseConnection(null);
         }
         private void AddExternalMetaDataColumn(IDTSInput100 input, IDTSInputColumn100 inputColumn)
         {
             // Set the properties of the external metadata column.
             IDTSExternalMetadataColumn100 externalColumn = input.ExternalMetadataColumnCollection.New();
             externalColumn.Name = inputColumn.Name;
             externalColumn.Precision = inputColumn.Precision;
             externalColumn.Length = inputColumn.Length;
             externalColumn.DataType = inputColumn.DataType;
             externalColumn.Scale = inputColumn.Scale;

             // Map the external column to the input column.
             inputColumn.ExternalMetadataColumnID = externalColumn.ID;
         }
         public override DTSValidationStatus Validate()
         {
             bool pbCancel = false;

             // Validate that there is one input.
             if (ComponentMetaData.InputCollection.Count != 1)
             {
                 ComponentMetaData.FireError(0, ComponentMetaData.Name, "Incorrect number of inputs.", "", 0, out pbCancel);
                 return DTSValidationStatus.VS_ISCORRUPT;
             }

              // validate that there is one connection manager:
             if (ComponentMetaData.RuntimeConnectionCollection.Count != 1)
             {
                 ComponentMetaData.FireError(0, ComponentMetaData.Name, "Incorrect number of Connection Managers.", "", 0, out pbCancel);
                 return DTSValidationStatus.VS_ISBROKEN;

             }

             // Let the base class verify that the input column reflects the output 
             // of the upstream component.
             DTSValidationStatus retval = base.Validate();

             if (retval != DTSValidationStatus.VS_ISVALID)
                 return retval;

             // if here, we have input columns and are valid. Now, create external metadata columns:

             foreach (IDTSInputColumn100 inputCol in ComponentMetaData.InputCollection[0].InputColumnCollection)
             {
                 AddExternalMetaDataColumn(ComponentMetaData.InputCollection[0], inputCol);
             }
             return retval;
         }

         int curFile;
         int rowsInCurrentFile = 1;
         System.IO.Packaging.Package dataFile;
         System.IO.Packaging.PackagePart dataPart;
         String tableName;
         XmlDocument dom;
         DataGetter [] columnGetters;
         IDTSConnectionManager100 Connection;
         System.IO.BinaryWriter writeStream;


         private class DataGetter
         {
             public String colName;
             public int colLength;
             public int colIndex;
             public int precision;
             public int scale;
             public System.IO.BinaryWriter binWriter;
             public System.IO.Packaging.Package dataFile;
             public CompressedFileDest parent;
             public long   rowsInCurrentFile = 0;

             public virtual void ProcessData(PipelineBuffer buf)
             {

             }

             public void processNewFile(ref System.IO.Packaging.Package newDataFile, ref System.IO.BinaryWriter newBinWriter)
             {
                 dataFile = newDataFile;
                 binWriter = newBinWriter;
                 rowsInCurrentFile = 0;
             }
         }
         private class Int16Getter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetInt16(colIndex));
                 }
             }
         }

         private class Int32Getter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetInt32(colIndex));
                 }
             }
         }
         private class Int64Getter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetInt64(colIndex));
                 }
             }
         }
         private class SByteGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetSByte(colIndex));
                 }
             }
         }
         private class SingleGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetSingle(colIndex));
                 }
             }
         }
         private class StringGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     //String needs ToString contain as length -- maybe not. Testing it without.
                     //binWriter.Write(buf.GetString(colIndex).Length);
                     binWriter.Write(buf.GetString(colIndex));
                 }
             }
         }
         private class TimeGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetTime(colIndex).Ticks);
                 }
             }
         }
         private class UInt16Getter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetUInt16(colIndex));
                 }
             }
         }
         private class UInt32Getter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetUInt32(colIndex));
                 }
             }
         }
         private class UInt64Getter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetUInt64(colIndex));
                 }
             }
         }
         private class GuidGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetGuid(colIndex).ToString());
                 }
             }
         }
         private class DoubleGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                    binWriter.Write(buf.GetDouble(colIndex));
                 }
             }
         }
         private class DecimalGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetDecimal(colIndex));
                 }
             }
         }
         // I'm not sure yet if the DateTimeOffset is really working right
         private class DateTimeOffsetGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     DateTimeOffset dto = buf.GetDateTimeOffset(colIndex);
                     binWriter.Write(dto.Ticks);
                     binWriter.Write(dto.DateTime.Ticks);
                 }
             }
         }
         private class DateTimeGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetDateTime(colIndex).Ticks);
                 }
             }
         }
         private class DateGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                if (buf.IsNull(colIndex))
                {
                    binWriter.Write(true);
                }
                else
                {
                    binWriter.Write(false);
                    binWriter.Write(buf.GetDate(colIndex).Ticks);
                }
             }
         }

         private class BytesGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     // bytes need a length:
                     binWriter.Write(buf.GetBytes(colIndex).Length);
                     binWriter.Write(buf.GetBytes(colIndex));
                 }
             }
         }
         private class ByteGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetByte(colIndex));
                 }
             }
         }
         private class BooleanGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write(true);
                 }
                 else
                 {
                     binWriter.Write(false);
                     binWriter.Write(buf.GetBoolean(colIndex));
                 }
             }
         }
         private class TextGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write((byte)1);
                 }
                 else
                 {
                     binWriter.Write(false);
                     offset = 0;
                     blobSize = buf.GetBlobLength(colIndex);
                     binWriter.Write(blobSize);
                     byte[] readBuf = new byte[bytesInArray];
                     Int16 bufCounter = 0;  // used to flush and prevent the isolated storage error (I hope)
                     while (offset < blobSize)
                     {
                         readBuf = buf.GetBlobData(colIndex, offset, bytesInArray);
                         binWriter.Write(readBuf, 0, readBuf.Length);
                         //binWriter.Write(readBuf);
                         offset += bytesInArray;
                         if (bufCounter++ > 60)
                         {
                             binWriter.Flush(); // flushing before Packing hits 8MB prevents the IsolatedStorage error
                             bufCounter = 0;
                         }
                     }
                    // always flush at the end of writing blob data - this prevents the IsolatedStorage error
                     binWriter.Flush();
                 }
             }
             uint blobSize;
             int bytesInArray = 1024;
             int offset = 0;
         }
         private class ImageGetter : CompressedFileDest.DataGetter
         {
             public override void ProcessData(PipelineBuffer buf)
             {
                 if (buf.IsNull(colIndex))
                 {
                     binWriter.Write((byte)1);
                 }
                 else
                 {
                     binWriter.Write(false);
                     offset = 0;
                     blobSize = buf.GetBlobLength(colIndex);
                     binWriter.Write(blobSize);
                     byte[] readBuf = new byte[bytesInArray];
                     Int16 bufCounter = 0;  // used to flush and prevent the isolated storage error (I hope)
                     while (offset < blobSize)
                     {
                         readBuf = buf.GetBlobData(colIndex, offset, bytesInArray);
                         binWriter.Write(readBuf, 0, readBuf.Length);
                         // binWriter.Write(readBuf);
                         offset += bytesInArray;
                         if (bufCounter++ > 60)
                         {
                             binWriter.Flush();
                             bufCounter = 0;
                         }
                     }
                     // always flush at the end of writing blob data. This prevents the isolated storage error:
                     binWriter.Flush();
  
                 }
                 rowsInCurrentFile++;

             }
             uint blobSize;
             int bytesInArray = 1024;
             int offset = 0;
         }

     }

}
